package cn.me.alphamq.nameserver.event.spi.listener.service;

import cn.me.alphamq.common.entity.req.ChangeMasterReq;
import cn.me.alphamq.common.entity.req.Shift2MasterReq;
import cn.me.alphamq.common.enumeration.BrokerRoleEnum;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.enumeration.ServiceInstanceTypeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.store.StopWatchManager;
import cn.me.alphamq.common.sync.StopWatch;
import cn.me.alphamq.common.util.AttrUtil;
import cn.me.alphamq.common.util.NettyUtil;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.nameserver.event.model.service.UnRegistryEvent;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.nameserver.store.ReplicationMsgQueueManager;
import cn.me.alphamq.nameserver.store.ServiceInstance;
import cn.me.alphamq.nameserver.store.ServiceInstanceManager;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 服务的注销事件的监听器，注销事件是在连接 inactive 之后被 nameserver 检测到，并投递到事件总线中
 *
 * @author f
 */
public class UnRegistryListener implements Listener<UnRegistryEvent> {

    private static final Logger log = LoggerFactory.getLogger(UnRegistryListener.class);

    private final ServiceInstanceManager serviceInstanceManager = Container.getServiceInstanceManager();
    private final ReplicationMsgQueueManager replicationMsgQueueManager = Container.getReplicationMsgQueueManager();

    /**
     * 停表
     */
    private final StopWatch stopWatch = StopWatchManager.newStopWatch();

    // 本方法负责将指定服务实例从 ServiceInstanceManager 的表中移除
    @Override
    public void onReceive(UnRegistryEvent event) throws Exception {
        ChannelHandlerContext ctx = event.getCtx();

        // 如果服务实例的标识为空，则报错
        String ctxId = NettyUtil.getCtxId(ctx);
        if (ctxId == null) {
            ctx.close();
            throw new RuntimeException("incorrect username or password");
        }

        // 移除服务实例
        ServiceInstance serviceInstance = serviceInstanceManager.remove(ctxId);
        String role = AttrUtil.getRoleFromAttrs(serviceInstance.getAttrs());
        String group = AttrUtil.getGroupFromAttrs(serviceInstance.getAttrs());

        // 如果是主节点，则进行主从切换
        if (BrokerRoleEnum.MASTER.getCode().equals(role)) {
            List<ServiceInstance> slaveList = serviceInstanceManager.getServiceInstanceMap().values().stream()
                    .filter(si -> group.equals(AttrUtil.getGroupFromAttrs(si.getAttrs())))
                    .sorted((si1, si2) -> Long.compare(
                            AttrUtil.getVersionFromAttrs(si2.getAttrs()),
                            AttrUtil.getVersionFromAttrs(si1.getAttrs())
                    )).collect(Collectors.toList());
            if (!slaveList.isEmpty()) {
                // 变更 新主节点 的 attrs 和 其 broker 内部的配置
                ServiceInstance newMaster = slaveList.get(0);
                AttrUtil.putRoleToAttrs(newMaster.getAttrs(), BrokerRoleEnum.MASTER.getCode());
                newMaster.getCtx().writeAndFlush(new TcpMsg(EventCodeEnum.SHIFT_TO_MASTER_REQ,
                        JSON.toJSONBytes(new Shift2MasterReq())));

                // 让其它 nameserver 节点知道主节点变更了
                replicationMsgQueueManager.putShiftyServiceInstance(newMaster, stopWatch);

                // 将 新主节点 的地址传递给其它 从节点，让其与之建立连接
                TcpMsg changeMasterReq = new TcpMsg(EventCodeEnum.CHANGE_MASTER_REQ, JSON.toJSONBytes(
                        new ChangeMasterReq(newMaster.getIp() + ":" + newMaster.getPort())));
                for (int i = 1; i < slaveList.size(); i++) {
                    ChannelHandlerContext slaveCtx = slaveList.get(i).getCtx();
                    if (slaveCtx != null) {
                        slaveCtx.writeAndFlush(changeMasterReq);
                    }
                }
            }
        }

        // 关闭通道
        ctx.close();
    }
}
